package io.kiki.sba.registry.remoting.netty;


import io.kiki.sba.registry.api.ChannelHandler;
import io.kiki.sba.registry.api.Client;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.util.OsUtils;
import io.kiki.stack.netty.config.Configs;
import org.apache.commons.lang.StringUtils;

import java.util.Arrays;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


public class NettyDataExchanger implements DataExchanger<ChannelHandler> {

    static {
        // def size=400, it is too big, and queueSize is too small
        System.setProperty(Configs.TP_QUEUE_SIZE, String.valueOf(1000 * 10));
        System.setProperty(Configs.TP_MIN_SIZE, String.valueOf(OsUtils.getCpuCount() * 10));
        System.setProperty(Configs.TP_MAX_SIZE, String.valueOf(OsUtils.getCpuCount() * 10));

        // bolt.tcp.heartbeat.maxtimes * bolt.tcp.heartbeat.interval
        // If the client has not sent heartbeat for 45 seconds, it is determined as down
        // The default value of 90s is too long. Change it to 45s to reduce the impact of physical
        // machine downtime
        if (StringUtils.isEmpty(System.getProperty(Configs.TCP_SERVER_IDLE))) {
            System.setProperty(Configs.TCP_SERVER_IDLE, String.valueOf(45000));
        }
    }

    private final Map<String, Client> serverTypeToClientMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<Integer, io.kiki.sba.registry.api.Server> portToServerMap = new ConcurrentHashMap<>();

    @Override
    public Client connect(String serverType, Url serverUrl, ChannelHandler... channelHandlers) {
        return this.connect(serverType, serverUrl, 1, channelHandlers);
    }

    @Override
    public Client connect(String serverType, Url serverUrl, int connNum, ChannelHandler... channelHandlers) {
        if (channelHandlers == null) {
            throw new IllegalArgumentException("channelHandlers cannot be null!");
        }
        Client client = serverTypeToClientMap.computeIfAbsent(serverType, key -> newNettyClient(connNum, channelHandlers));
        // if serverUrl is null, just init the client
        // the conn will auto create if absent
        if (serverUrl != null) {
            client.connect(serverUrl);
        }
        return client;
    }

    private NettyClient newNettyClient(int connNum, ChannelHandler[] channelHandlers) {
        NettyClient nettyClient = new NettyClient(connNum);
        nettyClient.initHandlers(Arrays.asList(channelHandlers));
        return nettyClient;
    }

    @Override
    public Client getClient(String serverType) {
        return serverTypeToClientMap.get(serverType);
    }

    @Override
    public io.kiki.sba.registry.api.Server open(Url url, int lowWaterMark, int highWaterMark, ChannelHandler... channelHandlers) {
        Server server = createServer(url, channelHandlers);
        server.configWaterMark(lowWaterMark, highWaterMark);
        server.startServer();
        return server;
    }

    @Override
    public io.kiki.sba.registry.api.Server open(Url url, ChannelHandler... channelHandlers) {
        Server server = createServer(url, channelHandlers);
        server.startServer();
        return server;
    }

    private Server createServer(Url url, ChannelHandler... channelHandlers) {
        if (channelHandlers == null) {
            throw new IllegalArgumentException("channelHandlers cannot be null!");
        }
        Server server = createBoltServer(url, channelHandlers);
        setServer(server, url);
        return server;
    }

    protected Server createBoltServer(Url url, ChannelHandler[] channelHandlers) {
        return new Server(url, Arrays.asList(channelHandlers));
    }


    /**
     * add server into serverMap
     */
    public void setServer(io.kiki.sba.registry.api.Server server, Url url) {
        portToServerMap.putIfAbsent(url.getPort(), server);
    }


    @Override
    public io.kiki.sba.registry.api.Server getServer(Integer port) {
        return portToServerMap.get(port);
    }


}
